package com.data.hijack.service;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.net.ssl.SSLException;

import org.apache.log4j.Logger;

import com.data.hijack.constant.Constants;
import com.data.hijack.context.HolderContext;
import com.data.hijack.http.HttpConstants;
import com.data.hijack.http.ResponseException;
import com.data.hijack.http.Status;
import com.data.hijack.protocol.executor.AbstractSchemaExecutor;
import com.data.hijack.readable.ReadablePlugin;
import com.data.hijack.util.AppUtils;
import com.data.hijack.util.Asserts;
import com.data.hijack.util.BeanUtils;
import com.data.hijack.util.StringUtils;

import cn.hutool.core.thread.ExecutorBuilder;

/**
 *
 * @author husq
 */
public class MessageHandler implements Handler, Runnable {
    private static final Logger logger = Logger.getLogger(MessageHandler.class);

    private static final Object READABLE_PLUGIN_REGISTRY = HolderContext.get(Constants.READABLE_PLUGIN_REGISTRY);
    private static final Map<String, Object> CONFIG_CONTEXT =
        (Map<String, Object>)HolderContext.get(Constants.CONFIG_CONTEXT);
    private static final String CORE_COUNT = (String)CONFIG_CONTEXT.get(Constants.HANDLER_CORE_COUNT);
    private static final String MAX_COUNT = (String)CONFIG_CONTEXT.get(Constants.HANDLER_MAX_COUNT);
    private static int maxCount = Integer.parseInt(MAX_COUNT);
    private static ExecutorService executor =
        ExecutorBuilder.create().setCorePoolSize(Integer.parseInt(CORE_COUNT)).setMaxPoolSize(maxCount)
            .setKeepAliveTime(60, TimeUnit.SECONDS).setWorkQueue(new LinkedBlockingQueue<>(maxCount)).build();
    private static final String TIME_OUT = "30000";
    private static final String TIME_WAIT = "60000";
    private volatile long timeWait;
    private String bizNameInPlug;
    private SysConfig consts = new SysConfig();
    private Selector selector;
    private SocketChannel socketChannel;

    public MessageHandler(SocketChannel socketChannel, int ops, String bizName) {
        this.socketChannel = socketChannel;
        this.consts.setBizName(bizName);

        try {
            this.selector = Selector.open();
            this.socketChannel.configureBlocking(false);
            SelectionKey selectionKey = this.socketChannel.register(this.selector, ops);
            this.selector.wakeup();
        } catch (IOException e) {
            logger.error(e);
        } catch (Exception e) {
            logger.error(e);
        }
    }

    @Override
    public void run() {
        HolderContext.putOnlyPair(Constants.CONFIG_CONTEXT, CONFIG_CONTEXT);
        HolderContext.putOnlyPair(Constants.READABLE_PLUGIN_REGISTRY, READABLE_PLUGIN_REGISTRY);

        try {
            this.init();
            while (true) {
                long start = System.currentTimeMillis();
                int number = this.selector.select(this.timeWait);
                if (this.timeWait(start) && number == 0) {
                    break;
                }

                Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    this.handleSocket(selectionKey);
                }
            }
        } catch (IOException e) {
            this.errResp(e);
        } catch (ClosedSelectorException e) {
            //
        } catch (Exception e) {
            this.errResp(e);
        } finally {
            HolderContext.expunge();
            AppUtils.closeChannel(this.socketChannel, this.selector);
        }
    }

    private boolean timeWait(long s) {
        long e = System.currentTimeMillis();
        long d = e - s;
        if (d >= this.timeWait) {
            return true;
        }

        return false;
    }

    private void handleSocket(SelectionKey selectionKey) throws Exception {
        String executorStr = Constants.EXECUTORS + Constants.DOT + this.consts.getSchema();
        AbstractSchemaExecutor executor = this.getExecutor(executorStr);
        executor.execute(this.selector, selectionKey, this.consts);
    }

    private ReadablePlugin getPlugin(String bizName) {
        Map<String, Class<ReadablePlugin>> pluginMap =
            (Map<String, Class<ReadablePlugin>>)HolderContext.get(Constants.READABLE_PLUGIN_REGISTRY);
        Class<ReadablePlugin> readablePluginClass = pluginMap.get(bizName);

        ReadablePlugin readablePlugin = null;
        try {
            readablePlugin = BeanUtils.instantiateClass(readablePluginClass);
        } catch (Exception e) {
            logger.error("ReadablePlugin", e);
            return null;
        }

        HolderContext.putOnlyPair(Constants.READABLE_PLUGIN_INSTANCE, readablePlugin);

        return readablePlugin;
    }

    private Map<String, String> getSysConfig(String gettingBizName) throws Exception {
        Map<String, Object> configContext = (Map<String, Object>)HolderContext.get(Constants.CONFIG_CONTEXT);
        Map<String, String> bizConfig = (Map<String, String>)configContext.get(gettingBizName);

        Asserts.notNull(bizConfig, "bizConfig");
        Asserts.check(!bizConfig.isEmpty(), "Obtain system configuration is empty");

        return bizConfig;
    }

    private String getErrReply(Map<String, String> sysConfig) {
        String errReply = AppUtils.getValue(this.bizNameInPlug, Constants.ERR_REPLY, sysConfig);

        if (StringUtils.isBlank(errReply)) {
            errReply = "[EDH001]Request processing error!";
        }

        if (Objects.equals(this.consts.getStatus(), Constants.STATUS_0)) {
            errReply = "[WDH001]Unauthorized access.";
        }

        return errReply;
    }

    private void init() throws Exception {
        ReadablePlugin readablePlugin = getPlugin(this.consts.getBizName());;
        if (Objects.isNull(readablePlugin)) {
            this.bizNameInPlug = this.consts.getBizName();
        } else {
            this.bizNameInPlug = readablePlugin.getBizName(this.socketChannel);
        }
        this.consts.setSysConfig(getSysConfig(this.bizNameInPlug));
        this.setValue(this.consts.getSysConfig());

        Asserts.check(!Objects.equals(this.consts.getStatus(), Constants.STATUS_0),
            "The current service has been shut down[%s]", this.bizNameInPlug);

        Asserts.notEmpty(this.bizNameInPlug, "bizName");

        HolderContext.putOnlyPair(Constants.BIZ_NAME_IN_PLUGIN, this.bizNameInPlug);
    }

    private void errResp(Exception e) {
        final String comment = AppUtils.getValue(this.bizNameInPlug, Constants.COMMENT, this.consts.getSysConfig());
        MessageHandler.logger.error(comment, e);
        String errReply = this.getErrReply(this.consts.getSysConfig());
        String cst = this.consts.getCharset();
        if (StringUtils.isBlank(cst)) {
            cst = Constants.DEFAULT_CHARSET;
        }
        Exception re = e;
        if (StringUtils.isNotBlank(errReply)) {
            re = new ResponseException(Status.INTERNAL_ERROR, "SYSTEM ERROR");
        } else {
            errReply = e.getMessage();
        }

        try {
            if (consts.getSchema().equalsIgnoreCase(Constants.SCHEMA_HTTP)
                || consts.getSchema().equalsIgnoreCase(Constants.SCHEMA_HTTPS)) {
                this.httpErrResp(re, errReply);
                return;
            }

            ByteBuffer writeBuffer = Charset.forName(cst).encode(errReply);
            this.socketChannel.write(writeBuffer);
        } catch (IOException ioe) {
            logger.error(ioe);
        } catch (Exception ex) {
            logger.error(ex);
        }
    }

    private void httpErrResp(Exception e, String errInfo) throws UnsupportedEncodingException {
        String desc = "", message = "";
        int len = 0;
        if (e instanceof SocketException) {
            if (HttpConstants.STREAM_SHUTDOWN.equals(e.getMessage())) {
                throw new RuntimeException("Stream Shutdown");
            }
        } else if (e instanceof SocketTimeoutException) {
            logger.error("Request Timeout", e);
            desc = Status.REQUEST_TIMEOUT.getDescription();
            message = "Request Timeout";
        } else if (e instanceof SSLException) {
            logger.error("SSL PROTOCOL FAILURE", e);

            desc = Status.INTERNAL_ERROR.getDescription();
            message = "SSL PROTOCOL FAILURE";
        } else if (e instanceof IOException) {
            logger.error("SERVER INTERNAL ERROR: IOException", e);

            desc = Status.INTERNAL_ERROR.getDescription();
            message = "SERVER INTERNAL ERROR: IOException";
        } else if (e instanceof ResponseException) {
            ResponseException re = (ResponseException)e;
            logger.error(re.getStatus(), e);

            desc = re.getStatus().getDescription();
            message = errInfo;
        } else {
            logger.error("500 Internal Server Error", e);

            desc = Status.INTERNAL_ERROR.getDescription();
            message = "Internal Server Error";
        }

        if (StringUtils.isNotBlank(message)) {
            len = message.getBytes(consts.getCharset()).length;
        }
        String respStr = this.buildErrRespStr(desc, message, len);
        ByteBuffer writeBuffer = Charset.forName(consts.getCharset()).encode(respStr);

        try {
            this.socketChannel.write(writeBuffer);
        } catch (Exception ex) {
            logger.error(ex);
        }
    }

    private String buildErrRespStr(String desc, String message, int len) {
        SimpleDateFormat sdf = new SimpleDateFormat("E, d MMM yyyy HH:mm:ss 'GMT'", Locale.US);
        sdf.setTimeZone(TimeZone.getTimeZone("GMT"));

        StringBuilder respStr = new StringBuilder();
        respStr.append("HTTP/1.1 ").append(desc).append(" \r\n").append("Content-Type").append(": ")
            .append("text/plain;charset=").append(consts.getCharset()).append("\r\n").append("Date").append(": ")
            .append(sdf.format(new Date())).append("\r\n").append("content-length").append(": ").append(len)
            .append("\r\n").append("\r\n").append(message);

        return respStr.toString();
    }

    private void setValue(Map<String, String> sysConfig) {
        String charset = AppUtils.getValue(this.bizNameInPlug, Constants.CHARSET, sysConfig);
        if (StringUtils.isBlank(charset)) {
            charset = Constants.DEFAULT_CHARSET;
        }
        this.consts.setCharset(charset);
        this.consts.setStatus(AppUtils.getValue(this.bizNameInPlug, Constants.STATUS, sysConfig));
        this.consts.setMsgType(AppUtils.getValue(this.bizNameInPlug, Constants.MSG_TYPE, sysConfig));
        this.consts.setIsInterceptable(AppUtils.getValue(this.bizNameInPlug, Constants.IS_INTERCEPTABLE, sysConfig));
        this.consts.setIsTranslatable(AppUtils.getValue(this.bizNameInPlug, Constants.IS_TRANSLATABLE, sysConfig));
        this.consts.setIsChanged(AppUtils.getValue(this.bizNameInPlug, Constants.REMOTE_IS_CHANGED, sysConfig));
        this.consts.setHost(AppUtils.getValue(this.bizNameInPlug, Constants.REMOTE_URL_HOST, sysConfig));
        this.consts.setPort(AppUtils.getValue(this.bizNameInPlug, Constants.REMOTE_URL_PORT, sysConfig));
        this.consts.setPath(AppUtils.getValue(this.bizNameInPlug, Constants.REMOTE_URL_PATH, sysConfig));
        this.consts.setSchema(AppUtils.getValue(this.bizNameInPlug, Constants.SCHEMA, sysConfig));
        String timeout = AppUtils.getValue(this.bizNameInPlug, Constants.TIMEOUT, sysConfig);
        if (StringUtils.isBlank(timeout)) {
            timeout = TIME_OUT;
        }
        this.consts.setTimeout(timeout);
        String tw = AppUtils.getValue(this.bizNameInPlug, Constants.TIMEWAIT, sysConfig);
        if (StringUtils.isBlank(tw)) {
            tw = TIME_WAIT;
        }
        this.timeWait = Long.parseLong(tw);
        String lm = AppUtils.getValue(this.bizNameInPlug, Constants.LOGMESSAGE, sysConfig);
        boolean logMessage = StringUtils.isNotBlank(lm) ? Boolean.parseBoolean(lm.trim()) : false;
        this.consts.setLogMessage(logMessage);
        this.consts.setMode(AppUtils.getValue(this.bizNameInPlug, Constants.MODE, sysConfig));
        this.consts.setProccessor(AppUtils.getValue(this.bizNameInPlug, Constants.PROCCESSOR, sysConfig));
        String maxConn = AppUtils.getValue(this.bizNameInPlug, Constants.MAX_CONNECTIONS, sysConfig);
        if (StringUtils.isBlank(maxConn)) {
            maxConn = "0";
        }
        this.consts.setMaxConnections(Integer.parseInt(maxConn));
        this.consts.setTmpDir(AppUtils.getValue(this.bizNameInPlug, Constants.TMPDIR, sysConfig));
    }

    @Override
    public Object handle() {
        executor.execute(this);
        return null;
    }

    private AbstractSchemaExecutor getExecutor(String key) throws ClassNotFoundException {
        Map<String, Object> configContext = (Map<String, Object>)HolderContext.get(Constants.CONFIG_CONTEXT);
        String executorStr = (String)configContext.get(key);
        Asserts.notEmpty(executorStr, "Executor");

        Class<AbstractSchemaExecutor> klass = (Class<AbstractSchemaExecutor>)Class.forName(executorStr.trim());
        AbstractSchemaExecutor executor = BeanUtils.instantiateClass(klass);

        return executor;
    }
}
